package com.atguigu.day05;

import com.atguigu.bean.Event;
import com.atguigu.bean.WaterSensor;
import com.atguigu.day03.Flink01_Source_Customer;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;

public class Flink05_TimeWindow_TumblingWindow_AggFun_Process {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.通过自定义Source获取数据
        DataStreamSource<Event> streamSource = env.addSource(new Flink01_Source_Customer.ClickSource());

//        streamSource.print();

        //3.将相同url的数据聚合到一块
        KeyedStream<Event, String> keyedStream = streamSource.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event value) throws Exception {
                return value.url;
            }
        });

        //4.我们这里统计10秒钟的url浏览量，每5秒钟更新一次；  开启窗口大小为10s 滑动步长为5s 的滑动窗口
        WindowedStream<Event, String, TimeWindow> window = keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)));

        //5.同时使用增量聚合函数和全窗口函数
        window.aggregate(new AggregateFunction<Event, Integer, Integer>() {
                             @Override
                             public Integer createAccumulator() {
                                 return 0;
                             }

                             @Override
                             public Integer add(Event value, Integer accumulator) {
                                 return accumulator+1;
                             }

                             @Override
                             public Integer getResult(Integer accumulator) {
//                                 System.out.println(accumulator);
                                 return accumulator;
                             }

                             @Override
                             public Integer merge(Integer a, Integer b) {
                                 return null;
                             }
                         },
                //全窗口函数中的数据是由上面增量聚合函数最终返回的结果，因此迭代器中只有一条数据
                new ProcessWindowFunction<Integer, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {
                        //1.获取url的访问量
                        Integer count = elements.iterator().next();
//                        System.out.println(count);
                        out.collect("窗口：[" + context.window().getStart() / 1000 + "," + context.window().getEnd() / 1000 + ")" + "----->这个窗口中url的访问数是:" + count);
                    }
                }).print();


        env.execute();
    }
}
